package com.jintian.smart.kernel.switching.api;

import com.jintian.smart.kernel.common.spring.BeanSelfAware;
import com.jintian.smart.kernel.module.Model;
import com.jintian.smart.kernel.redis.lock.RedissonLock;
import com.jintian.smart.kernel.switching.dto.DataPacket;
import com.jintian.smart.kernel.switching.dto.FileDataPacket;
import com.jintian.smart.kernel.switching.dto.TransferMessage;
import com.jintian.smart.kernel.switching.entity.ETLExtractJob;
import com.jintian.smart.kernel.switching.mapper.IETLExtractJobMapper;
import io.netty.buffer.ByteBuf;
import jakarta.xml.bind.JAXBContext;
import lombok.extern.slf4j.Slf4j;
import net.lingala.zip4j.ZipFile;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.beetl.sql.core.mapping.StreamData;
import org.beetl.sql.core.query.LambdaQuery;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.io.Resource;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.nio.file.Path;
import java.time.LocalDateTime;
import java.util.*;

/**
 * 采集任务
 */
@Slf4j
public abstract class AbstractExtractorTask<Rule> implements IExtractTask<Rule>, BeanSelfAware, ApplicationContextAware {
    @Value("${mda.switching.rule-path}")
    private String rulePath;

    @Value("${mda.switching.extract.write-path}")
    private File writePath;

    @Value("${mda.switching.extract.retries:60}")
    private int retries = 60;

    @Value("${mda.switching.lock:false}")
    private boolean lock;

    @Value("${mda.switching.extract.sequence:true}")
    protected boolean sequence = true;

    @Autowired
    private IETLExtractJobMapper extractJobMapper;
    @Autowired(required = false)
    private IFileTransfer IFileTransfer;
    @Autowired(required = false)
    private RedissonLock redissonLock;

    private Class<Rule> ruleClass;

    @SuppressWarnings({"unchecked"})
    public AbstractExtractorTask() {
        Type genType = this.getClass().getGenericSuperclass();
        Type[] params = ((ParameterizedType) genType).getActualTypeArguments();
        this.ruleClass = (Class<Rule>) params[0];
    }

    protected File getDataTypeWriteDir() {
        return new File(writePath, this.getDataType().getName());
    }

    private ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    private AbstractExtractorTask<Rule> self;

    @SuppressWarnings({"unchecked"})
    @Override
    public void setSelf(Object proxyBean) {
        this.self = (AbstractExtractorTask<Rule>) proxyBean;
    }

    @Override
    public Rule loadRule(Resource resource) {
        // XStream xstream = new XStream();
        // Annotations.configureAliases(xstream, this.ruleClass);
        // try (InputStream in = resource.getInputStream()) {
        // Rule rule = this.ruleClass.cast(xstream.fromXML(in));
        // return rule;
        // } catch (Exception ex) {
        // throw new IllegalArgumentException(ex);
        // }
        try {
            JAXBContext context = JAXBContext.newInstance(this.ruleClass);
            try (InputStream in = resource.getInputStream()) {
                Rule rule = this.ruleClass.cast(context.createUnmarshaller().unmarshal(in));
                return rule;
            } catch (Exception ex) {
                throw new IllegalArgumentException(ex);
            }
        } catch (Exception e) {
            throw new IllegalArgumentException(e);
        }

    }

    @Override
    @Transactional
    public void execute() {
        // dataType = this.dataType order by ruleName,createTime
        LambdaQuery<ETLExtractJob> query = extractJobMapper.createLambdaQuery()
                .andEq(ETLExtractJob::getDataType, this.getDataType().getSimpleName())
                .order(ETLExtractJob::getRuleName, true).order(ETLExtractJob::getCreateTime, true);

        if (System.getProperty("switching-ruleNames") != null) {
            // and ruleName in ()
            query.andIn(ETLExtractJob::getRuleName,
                    Arrays.asList(System.getProperty("switching-ruleNames").split(",")));
        }
        // and (status = 'waiting' or status='error' and retries<60)
        query.and(query.condition().andEq(ETLExtractJob::getStatus, ETLExtractJob.Status.waiting.name())
                .or(query.condition().andEq(ETLExtractJob::getStatus, ETLExtractJob.Status.error.name())
                        .andLess(ETLExtractJob::getRetries, this.retries)));
        StreamData<ETLExtractJob> stream = query.stream();
        Set<String> noLockIgnoreRules = new HashSet<>();
        Set<String> sequenceErrorIgnoreRules = new HashSet<>();
        List<String> lockRules = new ArrayList<>();
        try {
            ETLExtractJob job;
            while ((job = stream.next()) != null) {
                if (this.sequence && sequenceErrorIgnoreRules.contains(job.getRuleName())) {
                    // 顺序执行、忽略sequenceErrorIgnoreRules
                    log.info("顺序执行，忽略：ETLExtractJob{}", job.getId());
                    continue;
                }
                if (noLockIgnoreRules.contains(job.getRuleName())) {
                    // 无规则锁忽略
                    log.info("无法获取分布式锁，忽略：ETLExtractJob{}", job.getId());
                    continue;
                }
                if (redissonLock != null && lock) {
                    if (redissonLock.lock(this.getClass().getName() + ":" + job.getRuleName(), 300)) {
                        // 记录锁的规则名，最后需释放
                        lockRules.add(job.getRuleName());
                    } else {
                        // 添加无锁忽略
                        noLockIgnoreRules.add(job.getRuleName());
                        log.info("无法获取分布式锁，忽略：ETLExtractJob{}", job.getId());
                        continue;
                    }
                }
                try {
                    self.execute(job);
                } catch (Exception ex) {
                }
                self.updateByJob(job);
                if (job.getStatus().equals(ETLExtractJob.Status.error.name())) {
                    if (this.sequence) {
                        // 如果顺序执行失败，忽略所有规则实例
                        sequenceErrorIgnoreRules.add(job.getRuleName());
                    }
                }
            }
        } finally {
            if (redissonLock != null) {
                for (String ruleName : lockRules) {
                    redissonLock.release(this.getClass().getName() + ":" + ruleName);
                }
            }
        }

    }

    protected File createOutputResourceDir(ETLExtractJob job) {
        File writeDir = new File(this.getDataTypeWriteDir(), job.getRuleName());
        if (!writeDir.exists()) {
            writeDir.mkdirs();
        }
        return writeDir;
    }

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void updateByJob(ETLExtractJob job) {
        this.extractJobMapper.updateById(job);
    }

    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public boolean execute(ETLExtractJob job) {
        File writeDir = this.createOutputResourceDir(job);
        try {
            Collection<Model> modelList = this.applicationContext.getBeansOfType(Model.class).values();
            Model model = null;
            Rule rule = null;
            for (Model m : modelList) {
                Resource resource = m.getModelContext().getResource("classpath:etl/" + this.rulePath + "/" + job.getRuleName() + ".xml");
                if (resource.exists()) {
                    rule = this.loadRule(resource);
                    model = m;
                    break;
                }
            }

            // 创建加载器
            Map<String, IExtractor> beans = model.getModelContext().getBeansOfType(IExtractor.class);
            boolean b = false;
            for (IExtractor IExtractor : beans.values()) {
                if (IExtractor.support(this.getDataType(), this.ruleClass, job.getRuleName())) {
                    b = true;
                    job.setExecuteTime(LocalDateTime.now());
                    File extractFile = IExtractor.extract(this.getDataType(), rule, job.getRuleName(), job.getVariables());
                    File outFile = null;
                    if (extractFile.length() > 100 * 1024 * 1024) {
                        FileDataPacket info = new FileDataPacket();
                        info.setDataType(this.getDataType().getName());
                        info.setRuleClass(this.ruleClass.getName());
                        info.setRuleName(job.getRuleName());
                        info.setFileName(extractFile.getName());
                        try (TransferMessage msg = TransferMessage.valueOf(info, null);) {
                            ByteBuf buf = msg.getContent();
                            byte[] dst = new byte[buf.readableBytes()];
                            buf.readBytes(dst);
                            File dmpFile = new File(writeDir, UUID.randomUUID().toString() + IExtractTask.TRANSFER_DMP_FILE_EXT);
                            FileUtils.writeByteArrayToFile(dmpFile, dst);
                            outFile = new File(writeDir, UUID.randomUUID().toString() + IExtractTask.TRANSFER_DZP_FILE_EXT);
                            ZipFile zip = new ZipFile(outFile);
                            try {
                                zip.addFile(extractFile);
                                zip.addFile(dmpFile);
                                zip.close();
                            } catch (Exception e) {
                                throw new IllegalArgumentException(e);
                            }
                            dmpFile.delete();
                        }
                    } else {
                        DataPacket info = new DataPacket();
                        info.setDataType(this.getDataType().getName());
                        info.setRuleClass(this.ruleClass.getName());
                        info.setRuleName(job.getRuleName());
                        try (InputStream stream = new FileInputStream(extractFile)) {
                            info.setData(IOUtils.toByteArray(stream));
                        }

                        try (TransferMessage msg = TransferMessage.valueOf(info, null);) {
                            ByteBuf buf = msg.getContent();
                            byte[] dst = new byte[buf.readableBytes()];
                            buf.readBytes(dst);
                            outFile = new File(writeDir, UUID.randomUUID().toString() + IExtractTask.TRANSFER_DMP_FILE_EXT);
                            FileUtils.writeByteArrayToFile(outFile, dst);
                        }
                    }
                    extractFile.delete();
                    if (IFileTransfer != null) {
                        IFileTransfer.putFile(outFile.getPath());
                    }
                    Path outputResource = outFile.toPath();
                    job.setOutputResource(outputResource.subpath(this.getDataTypeWriteDir().toPath().getNameCount(), outputResource.getNameCount()).toString());
                    job.setException(null);
                    job.setStatus(ETLExtractJob.Status.succeed.name());
                }
            }
            if (!b) {
                job.setStatus(ETLExtractJob.Status.error.name());
                job.setRetries(job.getRetries() + 1);
                job.setException("无匹配的采集器");
            }
            return b;
        } catch (Exception ex) {
            log.error("{}采集[id={}]失败", this.getDataType(), job.getId(), ex);
            job.setStatus(ETLExtractJob.Status.error.name());
            job.setRetries(job.getRetries() + 1);
            job.setException(ex.getMessage());
            return false;
        }
    }
}
